package com.ruoyi.system.service.impl;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import java.util.stream.Collectors;

import com.alibaba.fastjson2.JSON;
import com.ruoyi.common.config.AppConfig;
import com.ruoyi.common.constant.CacheConstants;
import com.ruoyi.common.core.redis.RedisCache;
import com.ruoyi.common.enums.message.MessageIsRead;
import com.ruoyi.common.enums.message.MessageIsSync;
import com.ruoyi.common.enums.message.MessageType;
import com.ruoyi.common.enums.rabbitmq.RabbitMqOpsFlag;
import com.ruoyi.common.utils.DateUtils;
import com.ruoyi.common.utils.StringUtils;
import com.ruoyi.common.utils.uuid.IdUtils;
import com.ruoyi.system.config.RabbitMQConfig;
import com.ruoyi.system.domain.vo.MessageVO;
import com.ruoyi.system.domain.vo.SessionVO;
import com.ruoyi.system.service.ISessionService;
import com.ruoyi.system.vo.RabbitMQInfoVO;
import com.ruoyi.system.webchat.WebChatServer;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RedissonClient;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.ruoyi.system.mapper.MessageMapper;
import com.ruoyi.system.domain.Message;
import com.ruoyi.system.service.IMessageService;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;

/**
 * 消息Service业务层处理
 *
 * @author ruoyi
 * @date 2025-09-12
 */
@Service
@Slf4j
public class MessageServiceImpl implements IMessageService {

    @Autowired
    private MessageMapper messageMapper;

    @Autowired
    private RedisCache redisCache;

    @Autowired
    private RedissonClient redissonClient; // 需引入 Redisson 依赖

    @Autowired
    private ISessionService sessionService;

    @Autowired
    private RabbitTemplate rabbitTemplate;


    /**
     * 当消息是未读的时候  设置redis缓存中的未读消息数
     *      参数：conversationId
     *      参数：curUserId
     *      参数：id
     * @param messageVO
     */
    @Override
    public void updateRedisUnReadInfo(MessageVO messageVO) {

        if(ObjectUtils.isEmpty(messageVO)){
            log.info("SessionServiceImpl.updateUnreadCount sessionVO is null");
            return;
        }

        if(StringUtils.isEmpty(messageVO.getConversationId()) || messageVO.getCurUserId() == null){
            log.info("SessionServiceImpl.updateUnreadCount 参数格式不正确，sessionID:{},userId：{}",messageVO.getId(),messageVO.getCurUserId());
            return;
        }

        // 更改redis中未读消息数
        String key = StringUtils.handleString(CacheConstants.USER_UNREAD_LIST, messageVO.getConversationId(), String.valueOf(messageVO.getCurUserId()));
        log.info("SessionServiceImpl.updateUnreadCount key:{}", key);

        List<String> messageIds = redisCache.getCacheObject(key);
        if(CollectionUtils.isEmpty(messageIds)){
            messageIds = new ArrayList<>();
            messageIds.add(messageVO.getId());
//            redisCache.setCacheObject(key, messageIds, AppConfig.CACHE_TTL, TimeUnit.MINUTES);
            redisCache.setCacheObject(key, messageIds, 1, TimeUnit.MINUTES);
        }else {
            messageIds.add(messageVO.getId());
//            redisCache.setCacheObject(key, messageIds, AppConfig.CACHE_TTL, TimeUnit.MINUTES);
            redisCache.setCacheObject(key, messageIds, 1, TimeUnit.MINUTES);
        }

    }

    /**
     * 获取当前用户所有的会话的所有消息列表
     * @param queryVO
     * @return
     */
    @Override
    public List<MessageVO> getAllMsgList(MessageVO queryVO) {
        List<MessageVO> result = new ArrayList<>();

        Set<MessageVO> redisMessageVOS = redisCache.getCacheSetByMatchKey(CacheConstants.MSG_KEY + "*");
        List<String> sessionIds = queryVO.getSessionIds();
        if(!CollectionUtils.isEmpty(redisMessageVOS)){
            // 缓存命中
            Set<Integer> collect = redisMessageVOS.stream().map(Message::getIsSync).collect(Collectors.toSet());
            List<String> messageIdKeys = redisMessageVOS.stream().map(item->{
                return CacheConstants.MSG_KEY + item.getId();
            }).collect(Collectors.toList());

            if(!collect.contains(MessageIsSync.NO.getCode())){
                // 如果全部都是同步的消息
                if(!CollectionUtils.isEmpty(sessionIds)){
                    HashSet<String> sessionIdSet = new HashSet<>(sessionIds);
                    result = redisMessageVOS.stream().filter(item -> sessionIdSet.contains(item.getConversationId())).collect(Collectors.toList());
                }else {
                    result = new ArrayList<>(redisMessageVOS);
                }
            }else {
                // 手动同步
                List<MessageVO> unSyncList = redisMessageVOS.stream().filter(item -> MessageIsSync.NO.getCode().equals(item.getIsSync())).collect(Collectors.toList());
                // 当redis中的未同步消息小于100条  不进行同步操作
                if(!CollectionUtils.isEmpty(unSyncList) && unSyncList.size() <= AppConfig.SYNC_MESSAGE_LIMIT_COUNT){
                    return new ArrayList<>(redisMessageVOS);
                }
                log.info("SessionServiceImpl.getAllMsgList 开始手动同步.....");
                Integer preMsgOpsVersion = redisCache.getCacheObject(CacheConstants.SYNC_OPS_VERSION);
                if(preMsgOpsVersion == null){
                    preMsgOpsVersion = 0;
                }

                for (MessageVO messageVO : unSyncList) {
                    messageVO.setIsSync(MessageIsSync.YES.getCode());
                }
                saveOrUpdateList(unSyncList);

                Integer curMsgOpsVersion = redisCache.getCacheObject(CacheConstants.SYNC_OPS_VERSION);
                if(!preMsgOpsVersion.equals(curMsgOpsVersion)){
                    Set<MessageVO> messageVOSet = redisCache.getCacheSetByMatchKey(CacheConstants.MSG_KEY + "*");
                    return new ArrayList<>(messageVOSet);
                }else {
                    // 删除缓存
                    redisCache.deleteObject(messageIdKeys);

                    // 重新从数据库中，获取到消息 并设置到redis中
                    result = this.queryMsgList(new MessageVO());

                    redisCache.setBatchObject(new HashSet<>(result),new Function<MessageVO, String>(){
                        @Override
                        public String apply(MessageVO messageVO) {
                            return CacheConstants.MSG_KEY + messageVO.getId();
                        }
                    });

                    redisCache.increment(CacheConstants.SYNC_OPS_VERSION,1);
                }
            }
            return result;
        }
        // 未命中缓存  从数据库中获取数据
        result = this.queryMsgList(new MessageVO());

        redisCache.setBatchObject(new HashSet<>(result),new Function<MessageVO, String>(){
            @Override
            public String apply(MessageVO messageVO) {
                return CacheConstants.MSG_KEY + messageVO.getId();
            }
        });

        return result;
    }

    /**
     * 获取用户未读消息列表
     *  查询参数
     *      sessionIds 会话ID集合  必传
     *      curUserId 当前用户ID  必传
     * @param queryVO 查询参数VO
     * @return 未读消息列表
     */
    @Override
    public Map<String, List<MessageVO>> getUnReadMsgList(MessageVO queryVO) {
        Map<String, List<MessageVO>> result  = new HashMap<>();

        if(queryVO.getCurUserId() == null){
            log.info("SessionServiceImpl.getUnReadMsgList 参数格式错误，curUserId:{}",queryVO.getCurUserId());
            return result;
        }
        Long curUserId = queryVO.getCurUserId();

        SessionVO sessionVO = new SessionVO();
        sessionVO.setUserId(curUserId);
        sessionVO.setOpsFunc(SessionVO.OpsFunc.getUnReadMsgList.getText());
        sessionVO.setId(queryVO.getConversationId());
        List<SessionVO> sessionVOS = null;
        try {
            sessionVOS = sessionService.querySessionList(sessionVO);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

        if(!CollectionUtils.isEmpty(sessionVOS)){
            List<String> sessionIds = sessionVOS.stream().map(item -> item.getId()).collect(Collectors.toList());
            queryVO.setSessionIds(sessionIds);

            List<MessageVO> allMessage = this.getAllMsgList(queryVO);
            List<MessageVO> messageUnReadList = allMessage.stream()
                    .filter(item -> MessageIsRead.NO.getCode().equals(item.getIsRead()) &&
                            item.getReceiverId() != null && item.getReceiverId().equals(curUserId)).collect(Collectors.toList());

            // messageUnReadList 根据sessionId 进行分组  key ：sessionId  value ：List<MessageVO>
            result = messageUnReadList.stream().collect(Collectors.groupingBy(Message::getConversationId));
        }
        return result;
    }


    /**
     * TODO 优化成：用户-会话的消息列表，在容器启动的时候，加载到redis中，之后的数据扭转，都是在redis中操作
     * @param queryVO
     * @return
     */
    @Override
    public List<MessageVO> queryCacheMsgList(MessageVO queryVO) {
        List<MessageVO> result = new ArrayList<>();

        String sessionId = queryVO.getConversationId();
        List<MessageVO> allMsgList = this.getAllMsgList(new MessageVO());

        if(!CollectionUtils.isEmpty(allMsgList)){
            allMsgList = allMsgList.stream().filter(item -> StringUtils.isNotEmpty(item.getConversationId()) && item.getConversationId().equals(sessionId))
                    .collect(Collectors.toList());

            // 按照发送时间排序 降序
            allMsgList = allMsgList.stream().sorted(Comparator.comparing(MessageVO::getSendTime)).collect(Collectors.toList());
            if (allMsgList.size() > AppConfig.CACHE_MSG_LIST_LIMIT_COUNT){
                log.info("MessageServiceImpl queryCacheMsgList 缓存消息数大于限制消息数，开始删除处理");

                // 删除多余的消息
                int curCount = 0;
                for (MessageVO messageVO : allMsgList) {
                    if(curCount <= AppConfig.CACHE_MSG_LIST_LIMIT_COUNT){
                        result.add(messageVO);
                    }
                }
            }else {
                result = allMsgList;
            }
        }
        return result;
    }

    /**
     * 设置消息已读
     * @param queryVO
     */
    @Override
    public void setMessageRead(MessageVO queryVO) {

        if(queryVO.getCurUserId() == null || StringUtils.isEmpty(queryVO.getConversationId())){
            log.info("MessageServiceImpl.setMessageRead 参数格式不正确，sessionID:{},userId：{}",queryVO.getConversationId(),queryVO.getCurUserId());
            return;
        }

        String conversationId = queryVO.getConversationId();
        Long userId =queryVO.getCurUserId();

        // 获取未读消息列表  1.用户在线未读
        String curUserUnreadKey = StringUtils.handleString(CacheConstants.USER_UNREAD_LIST, conversationId, String.valueOf(userId));
        // 2.用户离线未读---此时的数据是从MySQL中读取放入redis中
        Map<String, List<MessageVO>> unReadMsgList = this.getUnReadMsgList(queryVO);

        if(!CollectionUtils.isEmpty(unReadMsgList)){
            List<MessageVO> updateList = new ArrayList<>();
            List<MessageVO> insertList = new ArrayList<>();

            List<MessageVO> allMsgList = this.queryMsgList(new MessageVO());
            Map<String, MessageVO> allMsgMap = new HashMap<>();
            if(!CollectionUtils.isEmpty(allMsgList)){
                allMsgMap = allMsgList.stream().collect(Collectors.toMap(MessageVO::getId, item -> item));
            }

            List<MessageVO> messageVOS = unReadMsgList.get(conversationId);
            if(!CollectionUtils.isEmpty(messageVOS)){
                for (MessageVO messageVO : messageVOS) {
                    messageVO.setIsRead(MessageIsRead.YES.getCode());
                    messageVO.setIsSync(MessageIsSync.YES.getCode());

                    if(allMsgMap.containsKey(messageVO.getId())){
                        updateList.add(messageVO);
                    }else {
                        insertList.add(messageVO);
                    }
                }
            }
            saveBathEntity(insertList);
            updateBathByIds(updateList);

            // 删除 unreadList key
            redisCache.deleteObject(curUserUnreadKey);
            // 删除缓存 更换消息版本
            deleteAndChangeMsgVersionRedis(new HashSet<>());
        }
    }

    @Override
    public void saveOrUpdateList(List<MessageVO> messageVOList) {
        List<MessageVO> messageVOS = this.queryMsgList(null);

        List<MessageVO> updateMessageVOList = new ArrayList<>();
        List<MessageVO> insertMessageVOList = new ArrayList<>();

        if(!CollectionUtils.isEmpty(messageVOS)){
            // list  转化为  map  key是id  value是messageVO
            Map<String, MessageVO> messageVOMap = messageVOS.stream().collect(Collectors.toMap(Message::getId, item -> item));
            if(!CollectionUtils.isEmpty(messageVOMap)){
                for (MessageVO messageVO : messageVOList) {
                    if(messageVOMap.containsKey(messageVO.getId())){
                        updateMessageVOList.add(messageVO);
                    }else {
                        if(messageVO.getId() == null){
                            messageVO.setId(IdUtils.simpleUUID());
                        }
                        insertMessageVOList.add(messageVO);
                    }
                }
            }
        }
        saveBathEntity(insertMessageVOList);
        updateBathByIds(updateMessageVOList);
    }

    private void updateBathByIds(List<MessageVO> mysqlUnreadMessageList) {
        if(CollectionUtils.isEmpty(mysqlUnreadMessageList)){
            return;
        }
        this.messageMapper.updateBathByIds(mysqlUnreadMessageList);
    }


    @Override
    public void saveEntity(MessageVO messageVO) {
        insertMessage(messageVO);
    }

//    @Transactional(propagation = Propagation.REQUIRED)
    @Override
    public void saveBathEntity(List<MessageVO> messageVOList) {
        if(CollectionUtils.isEmpty(messageVOList)){
            return;
        }
        this.messageMapper.saveBathEntity(messageVOList);
    }



    @Override
    public List<MessageVO> queryMsgList(MessageVO messageVO) {

        List<MessageVO> result = this.messageMapper.queryMsgList(messageVO);
        handleMessageListEnum(result);
        return result;
    }

    private static void handleMessageListEnum(List<MessageVO> result) {

        for (MessageVO messageVO : result){
            handleMessageVO(messageVO);
        }

    }

    private static void handleMessageVO(MessageVO messageVO) {
        if(messageVO != null){
            if(MessageIsRead.YES.getCode().equals(messageVO.getIsRead())){
                messageVO.setIsReadStr(MessageIsRead.YES.getInfo());
            }else if(MessageIsRead.NO.getCode().equals(messageVO.getIsRead())){
                messageVO.setIsReadStr(MessageIsRead.NO.getInfo());
            }

            if(MessageType.TEXT.getCode().equals(messageVO.getMessageType())){
                messageVO.setMessageTypeStr(MessageType.TEXT.getInfo());
            }else if(MessageType.IMAGE.getCode().equals(messageVO.getMessageType())){
                messageVO.setMessageTypeStr(MessageType.IMAGE.getInfo());
            }else if(MessageType.FILE.getCode().equals(messageVO.getMessageType())){
                messageVO.setMessageTypeStr(MessageType.FILE.getInfo());
            }

            if(MessageIsRead.YES.getCode().equals(messageVO.getSenderStatus())){
                messageVO.setSenderStatusStr(MessageIsRead.YES.getInfo());
            }else if(MessageIsRead.NO.getCode().equals(messageVO.getSenderStatus())){
                messageVO.setSenderStatusStr(MessageIsRead.NO.getInfo());
            }

            if(MessageIsSync.NO.getCode().equals(messageVO.getIsSync())){
                messageVO.setIsSyncStr(MessageIsSync.NO.getInfo());
            }else if (MessageIsSync.YES.getCode().equals(messageVO.getIsSync())){
                messageVO.setIsSyncStr(MessageIsSync.YES.getInfo());
            }
        }
    }

    @Transactional(propagation = Propagation.REQUIRED,rollbackFor = RuntimeException.class)
    public void handleCacheSyncDb() {
        try {
            // 查询当前在线的用户
            Map<String, WebChatServer.WebSocketSessionInfo> cacheMap = redisCache.getCacheMap(CacheConstants.USER_ONLINE);
            log.info("TaskSchedulerConfig handleCacheSyncDb 当前在线用户数：{}", cacheMap);
            // TODO 待优化 查出当前时间产生的所有消息   目前查出的是整个系统中，所有用户产生的所有消息，如果这个数据量很大，是会出问题的

            Collection<MessageVO> allMessageVO = redisCache.getCacheSetByMatchKey(CacheConstants.MSG_KEY + "*");
            // allMessageVO 转化为  map  key是id  value是messageVO
            Map<String, MessageVO> allMessageVOMap = allMessageVO.stream().collect(Collectors.toMap(Message::getId, item -> item));

            Collection<MessageVO> allDbMessageVO = this.queryMsgList( null);
            Map<String, MessageVO> allDbMessageVOMap = allDbMessageVO.stream().collect(Collectors.toMap(Message::getId, item -> item));


            List<MessageVO> insertMessageVOList = new ArrayList<>();
            List<MessageVO> updateMessageVOList = new ArrayList<>();
            // allUserSessionMsgKeys 表示：所有用户产生的消息redis keys  USER_SESSION_CUR_MSG_LIST
            Set<String> allUserSessionMsgKeys = new HashSet<>();

            if(!CollectionUtils.isEmpty(cacheMap.entrySet())){
                for (Map.Entry<String, WebChatServer.WebSocketSessionInfo> sessionEntry : cacheMap.entrySet()) {

                    String curUserId = sessionEntry.getKey();
                    String versionKey = CacheConstants.UNREAD_MSG_VERSION + curUserId;
                    log.info("TaskSchedulerConfig handleCacheSyncDb 当前正在处理用户：{}", curUserId);

                    try {
                        String str = StringUtils.handleString(CacheConstants.USER_SESSION_CUR_MSG_LIST,null, curUserId);
                        Set<String> curUserSessionMsgKeys = redisCache.scanKeys(str, 1000);
                        allUserSessionMsgKeys.addAll(curUserSessionMsgKeys);
                        Map<String, Set<String>> messageCurMap = redisCache.rangeByScoreWithSet(curUserSessionMsgKeys, 1L, System.currentTimeMillis());
                        log.info("TaskSchedulerConfig handleCacheSyncDb 用户当前产生的聊天记录ids：{}", messageCurMap);

                        if (CollectionUtils.isEmpty(messageCurMap)){
                            continue;
                        }
                        HashSet<String> messageIdSet = new HashSet<>();
                        for (Set<String> value : messageCurMap.values()) {
                            messageIdSet.addAll(value);
//                        for (String strV : value) {
//                            messageIdKeySet.add(validMsgKey + strV);
//                        }
                        }

                        for (String messageId : messageIdSet) {
                            MessageVO messageVO = allMessageVOMap.get(messageId);
                            messageVO.setIsSync(MessageIsSync.YES.getCode());
                            if(allDbMessageVOMap.containsKey(messageId)){
                                // 更新
                                updateMessageVOList.add(messageVO);
                            }else {
                                //插入
                                insertMessageVOList.add(messageVO);
                            }
                        }

                        // TODO 删除redis中的数据  当前用户产生的消息列表   考虑需不需删除消息列表
//                    redisCache.removeBatchKeys(scanKeys);
//                    redisCache.removeBatchKeys(messageIdKeySet);
                        redisCache.increment(versionKey, 1);

                        log.info("TaskSchedulerConfig handleCacheSyncDb 删除消息列表 messageIdKeySet：{}", messageIdSet);
                    }catch (Exception e){
                        log.error("TaskSchedulerConfig handleCacheSyncDb 获取锁失败：{}", e.getMessage());
                    }
                }
            }else {
                // 拿所有的消息
                String str = StringUtils.handleString(CacheConstants.USER_SESSION_CUR_MSG_LIST,"*","*");
                Set<String> matchKey = redisCache.scanKeys(str, 1000);
                Map<String, Set<String>> messageCurMap = redisCache.rangeByScoreWithSet(matchKey, 1L, System.currentTimeMillis());


                if (!CollectionUtils.isEmpty(messageCurMap)){
                    HashSet<String> messageIdSet = new HashSet<>();
                    allUserSessionMsgKeys.addAll(matchKey);
                    for (Set<String> value : messageCurMap.values()) {
                        messageIdSet.addAll(value);
                    }

                    for (String messageId : messageIdSet) {
                        MessageVO messageVO = allMessageVOMap.get(messageId);
                        messageVO.setIsSync(MessageIsSync.YES.getCode());
                        if(allDbMessageVOMap.containsKey(messageId)){
                            // 更新
                            updateMessageVOList.add(messageVO);
                        }else {
                            //插入
                            insertMessageVOList.add(messageVO);
                        }
                    }
                }
            }

//            // 同步 未读消息
//            String unreadKeys = StringUtils.handleString(CacheConstants.USER_UNREAD_LIST,"*","*");
//            Set<List<String>> unreadIdList = redisCache.getCacheSetByMatchKey(unreadKeys);
//            if(!CollectionUtils.isEmpty(unreadIdList)){
//
//                List<String> allUnreadIdList = new ArrayList<>();
//                for (List<String> list : unreadIdList) {
//                    allUnreadIdList.addAll(list);
//                }
//
//                Set<MessageVO> allUnreadMessageVOList = new HashSet<>();
//                for (String messageId : allUnreadIdList) {
//                    if(allMessageVOMap.containsKey(messageId)){
//                        MessageVO messageVO = allMessageVOMap.get(messageId);
//                        messageVO.setIsSync(MessageIsSync.YES.getCode());
//                        allUnreadMessageVOList.add(messageVO);
//                    }
//                }
//
//                for (MessageVO messageVO : allUnreadMessageVOList) {
//                    if(allDbMessageVOMap.containsKey(messageVO.getId())){
//                        // 存在
//                        updateMessageVOList.add(messageVO);
//                    }else {
//                        // 不存在
//                        insertMessageVOList.add(messageVO);
//                    }
//                }
//            }

            if(!CollectionUtils.isEmpty(insertMessageVOList) || !CollectionUtils.isEmpty(updateMessageVOList)){
                this.saveBathEntity(insertMessageVOList);
                this.updateBathByIds(updateMessageVOList);
                TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronizationAdapter() {
                    @Override
                    public void afterCommit() {
                        deleteAndChangeMsgVersionRedis(allUserSessionMsgKeys);
                    }
                });

            }else {
                log.info("TaskSchedulerConfig handleCacheSyncDb 没有需要同步的数据");
            }
        }catch (Exception e){
            throw new RuntimeException(e.getMessage());
        }
    }

    private void deleteAndChangeMsgVersionRedis(Set<String> allUserSessionMsgKeys) {
        Integer preMsgOpsVersion = redisCache.getCacheObject(CacheConstants.SYNC_OPS_VERSION);
        if(preMsgOpsVersion == null){
            preMsgOpsVersion = 0;
        }

        List<MessageVO> allMessages = messageMapper.queryMsgList(null); // 从数据库中查全量数据  设置到redis中
        HashSet<MessageVO> messageVOList = new HashSet<>(allMessages);

        Integer curMsgOpsVersion = redisCache.getCacheObject(CacheConstants.SYNC_OPS_VERSION);

        if(!preMsgOpsVersion.equals(curMsgOpsVersion)){
            log.info("deleteAndChangeMsgVersionRedis 缓存版本不一致");
            return;
        }else {
            List<String> collect = allMessages.stream().map(Message::getId).collect(Collectors.toList());
            HashSet<String> messageIdKeySet = new HashSet<>();

            for (String str : collect) {
                messageIdKeySet.add(CacheConstants.MSG_KEY + str);
            }
            redisCache.deleteObject(messageIdKeySet);
            redisCache.setBatchObject(messageVOList, new Function<MessageVO, String>() {
                @Override
                public String apply(MessageVO messageVO) {
                    return CacheConstants.MSG_KEY + messageVO.getId();
                }
            });
            // 自增
            redisCache.increment(CacheConstants.SYNC_OPS_VERSION,1);
        }
        if(!CollectionUtils.isEmpty(messageVOList)){
            RabbitMQInfoVO rabbitMQInfoVO = new RabbitMQInfoVO();
            rabbitMQInfoVO.setOpsFlag(RabbitMqOpsFlag.SYNC_DB.getCode());
            rabbitMQInfoVO.setAllUserSessionMsgKeys(allUserSessionMsgKeys);
            rabbitTemplate.convertAndSend(RabbitMQConfig.CACHE_SYNC_DB_EXCHANGE, RabbitMQConfig.CACHE_ROUTING_DB_KEY, JSON.toJSONString(rabbitMQInfoVO));
        }
    }


    @Override
    public MessageVO selectMessageByStringId(String messageId) {
        return this.messageMapper.selectMessageByStringId(messageId);
    }


    /**
     * 查询消息
     *
     * @param id 消息主键
     * @return 消息
     */
    @Override
    public Message selectMessageById(Long id)
    {
        return messageMapper.selectMessageById(id);
    }

    /**
     * 查询消息列表
     *
     * @param message 消息
     * @return 消息
     */
    @Override
    public List<Message> selectMessageList(Message message)
    {
        return messageMapper.selectMessageList(message);
    }

    /**
     * 新增消息
     *
     * @param message 消息
     * @return 结果
     */
    @Override
    public int insertMessage(Message message) {
        message.setCreateTime(DateUtils.getNowDate());
        return messageMapper.insertMessage(message);
    }

    /**
     * 修改消息
     *
     * @param message 消息
     * @return 结果
     */
    @Override
    public int updateMessage(Message message) {
        message.setUpdateTime(DateUtils.getNowDate());
        return messageMapper.updateMessage(message);
    }

    /**
     * 批量删除消息
     *
     * @param ids 需要删除的消息主键
     * @return 结果
     */
    @Override
    public int deleteMessageByIds(Long[] ids)
    {
        return messageMapper.deleteMessageByIds(ids);
    }

    /**
     * 删除消息信息
     *
     * @param id 消息主键
     * @return 结果
     */
    @Override
    public int deleteMessageById(Long id)
    {
        return messageMapper.deleteMessageById(id);
    }


}
